package cn.itcast.task;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.StockIncrBean;
import cn.itcast.config.QuotConfig;
import cn.itcast.function.KeyFunction;
import cn.itcast.function.MinStockIncreWindowFunction;
import cn.itcast.inter.ProcessDataInterface;
import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011;

import java.util.Properties;

/**
 * 个股涨跌幅
 */
public class StockIncrTask implements ProcessDataInterface {

    @Override
    public void process(DataStream<CleanBean> waterData) {

        /**
         * 1.数据分组
         * 2.划分时间窗口
         * 3.创建bean对象
         * 4.分时数据处理（新建分时窗口函数）
         * 5.数据转换成字符串
         * 6.数据存储(单表)
         *  注意:数据插入到druid，单表存储沪深两市行情，基于分时级数据开发
         */
        Properties prop = new Properties();
        prop.setProperty("bootstrap.servers", QuotConfig.config.getProperty("bootstrap.servers"));
        FlinkKafkaProducer011<String> kafkaProducer = new FlinkKafkaProducer011<>(
                QuotConfig.config.getProperty("stock.increase.topic"),
                new SimpleStringSchema(),
                prop);

        //todo 1.数据分组
        waterData.keyBy(new KeyFunction())
                //todo 2.划分时间窗口
                .timeWindow(Time.seconds(60))
                //todo 4.分时数据处理（新建分时窗口函数）
                .apply(new MinStockIncreWindowFunction())//

                //todo 5.数据转换成字符串
                .map(new MapFunction<StockIncrBean, String>() {
                    @Override
                    public String map(StockIncrBean stockIncrBean) throws Exception {
                        return JSON.toJSONString(stockIncrBean);
                    }
                })
                .addSink(kafkaProducer);
        //todo 3.创建bean对象(StockIncrBean)
        //todo 6.数据存储(单表)

    }
}
